<?php

namespace app\components\queue\rabbitmq;

use Interop\Amqp\AmqpQueue;
use Interop\Amqp\AmqpTopic;
use Interop\Amqp\Impl\AmqpBind;
use yii\helpers\Json;
use yii\queue\ExecEvent;
use yii\queue\serializers\PhpSerializer;

class Queue extends \yii\queue\amqp_interop\Queue {

    /**
     * @var string 消息执行处理器
     */
    public $executeHandler;

    /**
     * Topic type
     * @var string
     */
    public $topicType;

    protected function setupBroker()
    {
        if ($this->setupBrokerDone) {
            return;
        }

        $queue = $this->context->createQueue($this->queueName);
        $queue->addFlag(AmqpQueue::FLAG_DURABLE);
        $queue->setArguments(['x-max-priority' => $this->maxPriority]);
        $this->context->declareQueue($queue);

        $topic = $this->context->createTopic($this->exchangeName);
        $topic->setType($this->topicType);
        $topic->addFlag(AmqpTopic::FLAG_DURABLE);
        $this->context->declareTopic($topic);

        $this->context->bind(new AmqpBind($queue, $topic));

        $this->setupBrokerDone = true;
    }

    /**
     * @param string $id of a job message
     * @param string $message
     * @param int $ttr time to reserve
     * @param int $attempt number
     * @return bool
     */
    protected function handleMessage($id, $message, $ttr, $attempt)
    {
        if(!empty($this->executeHandler)){
            if($this->serializer instanceof PhpSerializer){
                $job = unserialize($message);
            }else{
                $job = Json::decode($message);
            }
            $error = false;
        }else{
            list($job, $error) = $this->unserializeMessage($message);
        }
        $event = new ExecEvent([
            'id' => $id,
            'job' => $job,
            'ttr' => $ttr,
            'attempt' => $attempt,
            'error' => $error,
        ]);
        $this->trigger(self::EVENT_BEFORE_EXEC, $event);
        if ($event->handled) {
            return true;
        }
        if ($event->error) {
            return $this->handleError($event);
        }
        try {
            if(!empty($this->executeHandler)){
                $clazz = new $this->executeHandler();
                $event->result = $clazz->execute($this,$job);
            }else{
                $event->result = $event->job->execute($this);
            }
        } catch (\Exception $error) {
            $event->error = $error;
            return $this->handleError($event);
        } catch (\Throwable $error) {
            $event->error = $error;
            return $this->handleError($event);
        }
        $this->trigger(self::EVENT_AFTER_EXEC, $event);
        return true;
    }
}